package com.yanqu.road.utils.kafka;

import com.yanqu.road.utils.string.StringUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;

import java.util.Properties;

public class KafKaHelper {

    private static Producer<String, String> producer = new EmptyProducer();

    public static synchronized void initProducer(String servers){
        Properties properties = getConfigProperties(servers);
        if(properties != null) {
            Producer<String, String> tempProducer = new KafkaProducer<>(properties);
            if(producer != null){
                //放弃旧对象
                producer.flush();
                producer.close();
            }
            producer = tempProducer;
        }
    }

    public static Producer<String, String> getProducer(){
        return producer;
    }

    public static Properties getConfigProperties(String servers){
        Properties props = new Properties();
        if(StringUtils.isNullOrEmpty(servers)) {
            return null;
        }
        props.put("bootstrap.servers", servers); // Kafka 集群地址
        props.put("acks", "all"); // 生产者需要服务器的确认，all 表示所有副本都确认
        props.put("retries", 0); // 如果发送失败，重试次数
        props.put("batch.size", 16384); // 批次大小
        props.put("linger.ms", 1); // 发送延迟，以便批次填充
        props.put("buffer.memory", 33554432); // 生产者内存缓冲区大小
        props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 键的序列化器
        props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer"); // 值的序列化器
        return props;
    }
}
